package com.bingxu.flink.watermark;

import com.bingxu.flink.bean.WaterSensor;
import com.bingxu.flink.utils.TransFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/16 19:10
 * @modifier :
 */

public class WMEvent {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env
                .socketTextStream("hadoop102", 9999)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] fields = value.split(",");
                    return new WaterSensor(fields[0], Long.valueOf(fields[1]), Integer.valueOf(fields[2]));
                })
                // 分配时间戳和水印，参数指定对应的分配策略，静态方法对不同类型的流选择合适的分配器
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        // 设置时间分配器，一般是指定元素的那个字段作为时间戳
                        .withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (element, recordTimestamp) -> element.getTs()))
                .keyBy(ele -> ele.getId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<WaterSensor, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(WaterSensor value, Long accumulator) {

                        return accumulator + value.getVc();
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                }, new WindowFunction<Long, String, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window,
                                      Iterable<Long> input, Collector<String> out) throws Exception {
                        Long sum = input.iterator().next();
                        out.collect(s+" "+sum);
                    }
                })
                .print();

        env.execute();
    }
}
